package org.example;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.sql.Timestamp;
import java.util.Properties;

public class KafkaProducerDemo {
    private static final String TOPIC = "kevin-test";
    private static final String BOOTSTRAP_SERVERS = "10.150.11.74:9092";

    public static void main(String[] args) {
        // 1. 配置生产者参数
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 2. 创建生产者实例
        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            for (int i = 0; i < 10; i++) {
                // 3. 构造消息（JSON格式示例）
                String key = "key-" + (i+1);
                String name = "user-" + (i+1);
                Timestamp ts = new Timestamp(System.currentTimeMillis());

                UserSource userSource = new UserSource();
                userSource.setId(i+1);
                userSource.setName(name);
                userSource.setAge(10+i);
                userSource.setEventTime(ts);
/*                String value = String.format("{\"id\":%d,\"name\":\"user-%d\",\"ts\":%d}",
                    i, i, System.currentTimeMillis());*/

                // 4. 异步发送消息
                String value = JSON.toJSONString(userSource);
                producer.send(new ProducerRecord<>(TOPIC, key, value), (metadata, e) -> {
                    if (e == null) {
                        System.out.printf("Sent [key:%s, value:%s] to partition-%d offset-%d\n",
                            key, value, metadata.partition(), metadata.offset());
                    } else {
                        e.printStackTrace();
                    }
                });
            }
            producer.flush(); // 确保所有消息发送完成
        }
    }
}
